package plumber.coordinator;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import plumber.common.HasSetup;
import plumber.common.Utils;
import plumber.common.WorkItem;
import plumber.common.WorkItem.Status;

import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.services.simpledb.AmazonSimpleDBClient;
import com.amazonaws.services.simpledb.model.Attribute;
import com.amazonaws.services.simpledb.model.CreateDomainRequest;
import com.amazonaws.services.simpledb.model.Item;
import com.amazonaws.services.simpledb.model.PutAttributesRequest;
import com.amazonaws.services.simpledb.model.ReplaceableAttribute;
import com.amazonaws.services.simpledb.model.SelectRequest;
import com.amazonaws.services.simpledb.model.SelectResult;
import com.amazonaws.services.simpledb.model.UpdateCondition;

public class AmazonSimpleDBCoordinator extends AbstractPersistentCoordinator
		implements CoordinatorClient, CoordinatorServer, HasSetup {

    private static final String NULLZ = "NULLZ";
    private static final String LONG_FORMAT = "%012d";
    private static final String INT_FORMAT = "%06d";
    
    private AmazonSimpleDBClient sdb;
    private final String domain;
    
	public AmazonSimpleDBCoordinator(String domain, String endpoint, String key, String secret) {
        super();
        sdb = new AmazonSimpleDBClient(new BasicAWSCredentials(key, secret));
        sdb.setEndpoint(endpoint);
        this.domain = domain;
    }

    @Override
	public List<String> getCandidates(String namespace, int upto) {
        return this.contextConverter.toStorableStrings(makeWorkItems(query("select * from `" + domain + "` "  +
                " where " + STATUS_COLUMN + 
                " = '" + WorkItem.Status.AVAILABLE + "' and " +
                DATE_SCHEDULED_COLUMN + " < '" + String.format(LONG_FORMAT, System.currentTimeMillis()) + "'"
                + " and " + DATE_SCHEDULED_COLUMN + " is not null order by " + DATE_SCHEDULED_COLUMN        
        )));
        
	}

	private List<MyWorkItem> query(String sql) {
        return query(sql, Integer.MAX_VALUE);
    }
    
    private List<MyWorkItem> query(String sql, int upto) {
        List<MyWorkItem> myItems = new ArrayList<MyWorkItem>();

        SelectRequest select = new SelectRequest(sql);
        select.setConsistentRead(true);
        String nextT = null;
        while(true) {
            logger.info(String.format("query:%s next:%s", sql, nextT));
            select.setNextToken(nextT);
            SelectResult response = sdb.select(select);
            for(Item item : response.getItems()) {
                myItems.add(fromSdbItem(item));
                if(myItems.size() >= upto) {
                    return myItems;
                }
            }
            nextT = response.getNextToken();
            if(nextT == null) {
                return myItems;
            }
        }
    }

   

	private MyWorkItem fromSdbItem(Item item) {
        MyWorkItem result = new MyWorkItem();
        result.setId(item.getName());
        for(Attribute attr : item.getAttributes()) {
            String name = attr.getName();
            String value = attr.getValue();
            if(NULLZ.equals(value)) {
                value = null;
            }
            if(name.equals(CONTEXT_COLUMN)) {
                result.setContext(value);
            }
            if(name.equals(DATE_SCHEDULED_COLUMN)) {
                result.setScheduled(new Date(new Long(value)));
            }
            if(name.equals(FLOW_NAME_COLUMN)) {
                result.setFlowName(value);
            }
            if(name.equals(NAMESPACE_COLUMN)) {
                result.setNamespace(value);
            }
            if(name.equals(RID_COLUMN)) {
                result.setRequestId(value);
            }
            if(name.equals(STATUS_COLUMN)) {
                result.setStatus(Status.valueOf(value));
            }
            if(name.equals(STEP_ID_COLUMN)) {
                result.setFlowStepId(value);
            }
            if(name.equals(VERSION_COLUMN)) {
                result.setVersion(new Integer(value));
            }
            
        }
        return result;
    }        


    @Override
	public void advanceWork(String context, String oldItemId,
			String nextStepId, Status oldStatus, Date scheduled) {
        logger.info("advancing work: " + context + " "
                + oldStatus + " " + nextStepId);
        MyWorkItem oldItem = find(oldItemId);
        Utils.validatePresent(oldItem, "old item");
        if (nextStepId != null) {
            WorkItem newItem = makeNewItem(context, oldItem.getFlowName(), nextStepId, scheduled,
                    oldItem.getRequestId(), oldItem.getNamespace());

            if (!updateStatus(oldItemId, Status.IN_PROGRESS.toString(), 
                    oldStatus.toString(), context)) {
                throw new RuntimeException("was not abie to update status");
            }
            create(newItem);
        } else {
            updateStatus(oldItemId, Status.IN_PROGRESS.toString(),
                    oldStatus.toString(), context);
        }

	}

	@Override
	public List<String> getHistory(String requestId) {
	    return this.contextConverter.toStorableStrings(makeWorkItems(query("select * from `" + domain
                + "` where " + RID_COLUMN + " = '" + requestId + "'"
                + " and " + DATE_SCHEDULED_COLUMN + " is not null order by " + DATE_SCHEDULED_COLUMN        
	    )));
	}

	

	@Override
	public List<String> getAll(String namespace, int upto) {
	    return this.contextConverter.toStorableStrings(makeWorkItems(query("select * from `" + domain
                + "` where " + NAMESPACE_COLUMN + " = '" + namespace + "' "
                + " and " + DATE_SCHEDULED_COLUMN + " is not null order by " + DATE_SCHEDULED_COLUMN, upto)));
	}

	@Override
	public void create(WorkItem item) {
		List<ReplaceableAttribute> attrs = new ArrayList<ReplaceableAttribute>();
		attrs.add(new ReplaceableAttribute().withName(FLOW_NAME_COLUMN).withValue(nvl(item.getFlowName())));
		attrs.add(new ReplaceableAttribute().withName(NAMESPACE_COLUMN).withValue(nvl(item.getNamespace())));
		attrs.add(new ReplaceableAttribute().withName(CONTEXT_COLUMN).withValue(nvl(item.getContext())));
		attrs.add(new ReplaceableAttribute().withName(DATE_CREATED_COLUMN).withValue(nvl(String.format(LONG_FORMAT, item.getCreated().getTime()))));
        attrs.add(new ReplaceableAttribute().withName(DATE_SCHEDULED_COLUMN).withValue(nvl(String.format(LONG_FORMAT, item.getScheduled().getTime()))));
        attrs.add(new ReplaceableAttribute().withName(VERSION_COLUMN).withValue(nvl(String.format(INT_FORMAT, 0))));       
        attrs.add(new ReplaceableAttribute().withName(STEP_ID_COLUMN).withValue(nvl(item.getFlowStepId())));
        attrs.add(new ReplaceableAttribute().withName(RID_COLUMN).withValue(nvl(item.getRequestId())));
        attrs.add(new ReplaceableAttribute().withName(STATUS_COLUMN).withValue(nvl(item.getStatus().toString())));
        attrs.add(new ReplaceableAttribute().withName(PK_COLUMN).withValue(nvl(item.getId())));
        
        sdb.putAttributes(new PutAttributesRequest(domain, item.getId(), attrs));
	}

	private String nvl(String value) {
	    if(value == null) {
	        return NULLZ;
	    }
	    return value;
    }

    @Override
	public List<MyWorkItem> itemsMatchingRequestIdAndStatus(String requestId,
			String status) {
	    return query(
                "select * from `" + domain + "` where " + RID_COLUMN
                        + " = '" + requestId + "' and " + STATUS_COLUMN + " = '" + status + "'");
	}

	@SuppressWarnings("serial")
    @Override
	public boolean updateStatus(String id, String fromStatus, String status,
			String context) {
	    MyWorkItem item = find(id);
        if (item == null) {
            return false;
        }
        if(item.getStatus() != Status.valueOf(fromStatus)) {
            return false;
        }
        if(context == null) {
            context = item.getContext();
        }
        PutAttributesRequest put = new PutAttributesRequest();
        put.setDomainName(domain);
        final ReplaceableAttribute attr = new ReplaceableAttribute();
        attr.setName(STATUS_COLUMN);
        attr.setReplace(true);
        attr.setValue(status);
        final ReplaceableAttribute attr2 = new ReplaceableAttribute();
        attr2.setName(VERSION_COLUMN);
        attr2.setReplace(true);
        attr2.setValue(String.format(INT_FORMAT, item.getVersion() + 1));
        final ReplaceableAttribute attr3 = new ReplaceableAttribute();
        attr3.setName(CONTEXT_COLUMN);
        attr3.setReplace(true);
        attr3.setValue(context);
        put.setAttributes(new ArrayList<ReplaceableAttribute>() {{ add(attr); add(attr2); add(attr3); }});
        put.setExpected(new UpdateCondition(VERSION_COLUMN, String.format(INT_FORMAT, item.getVersion()), true));
        put.setItemName(item.getId());
        try {
            sdb.putAttributes(put);
            return true;
        } catch (Exception ex) {
            logger.warn("updating", ex);            
            return false;
        }
	}

	private MyWorkItem find(String id) {
	    List<MyWorkItem> found = query("select * from `" + domain 
                + "` where " + PK_COLUMN + " = '" + id + "'");
        if (found.size() == 0) {
            return null;
        }
        if (found.size() > 1) {
            throw new RuntimeException("more than one item matched id");
        }
        return found.get(0);
	    
	}

    @Override
    public void setup() {
       sdb.createDomain(new CreateDomainRequest(domain));
        
    }

    @Override
    public void shutdown() {
        
        
    }
}
